package cassandra.client;
 
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

/**
 * 
 * @author Anderson Marinho
 *
 */

public class CassandraClient {
         
         private static CassandraClient instance = null;
         private TTransport tr = null;
         TProtocol proto = null;
         Cassandra.Client client = null;
         
         private String host = "127.0.0.1";
         
         
         public CassandraClient() {
        	 TSocket socket = new TSocket(host, 9160);
    		 tr = new TFramedTransport(socket);
    		 proto = new TBinaryProtocol(tr);
             client = new Cassandra.Client(proto);
             try {
				tr.open();
			} catch (TTransportException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
         }
         
         public static CassandraClient getInstance() {
             if(instance == null) {
                instance = new CassandraClient();
             }
             return instance;
          }
         
         public void close(){
        	 this.tr.close();
        	 instance = null;
         }
         
         
         public void insertColumn(String keySpace, String columnFamily, String keyID, String columnName, String columnValue){
        	 
        	 try{
                 
                 client.set_keyspace(keySpace);
                 

                 ByteBuffer userid = ByteBuffer.wrap(keyID.getBytes());
                 ColumnParent parent = new ColumnParent(columnFamily);
                 long timestamp = System.currentTimeMillis();
                 

                 Column column = new Column();
                 column.setName(columnName.getBytes("UTF8"));
                 column.setValue(columnValue.getBytes("UTF8"));
                 column.setTimestamp(timestamp);
                 

                 // insert data
                 client.insert(userid, parent, column, ConsistencyLevel.ANY);
                 
        	 } catch(Exception e){
        		 e.printStackTrace();
        	 }
        	 
         }
         
         public void insertColumns(String keySpace, String columnFamily, String keyID, Map<String, String> columns){
        	 for (String column : columns.keySet()) {
				insertColumn(keySpace, columnFamily, keyID, column, columns.get(column));
			}
         }
         
         public Column getColumn(String keySpace, String columnFamily, String KeyID, String columnName){
        	 
        	 try{
                 
                 client.set_keyspace(keySpace);
                 

                 ByteBuffer userid = ByteBuffer.wrap(KeyID.getBytes());
                 ColumnPath colpath = new ColumnPath("Users");
                 colpath.setColumn_family(columnFamily);
                 colpath.setColumn("first".getBytes("UTF8"));
                 
                 // read single column
                 Column col = client.get(userid, colpath,
                                 ConsistencyLevel.ANY).getColumn();
                 
                 return col;

                 
        	 } catch(Exception e){
        		 e.printStackTrace();
        	 }
        	 
        	 return null;
         }
         
         public List<ColumnOrSuperColumn> getColumns(String keySpace, String columnFamily, String KeyID){
        	 
        	 try{
                 
                 client.set_keyspace(keySpace);
                 

                 ByteBuffer userid = ByteBuffer.wrap(KeyID.getBytes());
                 ColumnParent parent = new ColumnParent(columnFamily);
                 
                 
                 SlicePredicate predicate = new SlicePredicate();
                 SliceRange sliceRange = new SliceRange();
                 sliceRange.setStart(new byte[0]);
                 sliceRange.setFinish(new byte[0]);
                 predicate.setSlice_range(sliceRange);
 
                 List<ColumnOrSuperColumn> columns = client.get_slice(userid,
                                 parent, predicate, ConsistencyLevel.ANY);
                 
                 return columns;

                 
        	 } catch(Exception e){
        		 e.printStackTrace();
        	 }
        	 
        	 return null;
         }
         
         
         public static void main(String[] args) throws Exception{
        	 
        	 
        	 //To seguindo o mesmo exemplo das instres do Cassandra.
        	 
        	 String keySpace = "Keyspace1";
             String columnFamily = "Users";
             String keyID = "andymarinho";
        	 
             
             //Inserindo uma coluna em uma linha da base
             CassandraClient.getInstance().insertColumn(keySpace, columnFamily, keyID, "first", "Anderson");
             
             //Inserindo colunas em uma linha da base
             Map<String, String> keyValueCols = new HashMap<String, String>();
             keyValueCols.put("last", "Marinho");
             keyValueCols.put("age", "29");
             CassandraClient.getInstance().insertColumns(keySpace, columnFamily, keyID, keyValueCols);
             
             
             
             //Buscando na base uma coluna (coluna "age")
        	 Column col = CassandraClient.getInstance().getColumn(keySpace, columnFamily, keyID, "age");
        	 
        	 System.out.println("column name: " + new String(col.getName(), "UTF8"));
             System.out.println("column value: " + new String(col.getValue(), "UTF8"));
             
             
             //Buscando na base todas as colunas de uma linha
             List<ColumnOrSuperColumn> cols = CassandraClient.getInstance().getColumns(keySpace, columnFamily, keyID);
        	 
             for (ColumnOrSuperColumn colOrSuperCol : cols) {
            	 
                 Column column = colOrSuperCol.column;
                 System.out.println(new String(column.getName(), "UTF8") + " -> "
                               + new String(column.getValue(), "UTF8"));
             }
             
             CassandraClient.getInstance().close();
             
         }
         
}